package top.zxybird.yun.serialization.thrift;

import java.io.UnsupportedEncodingException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import top.zxybird.yun.common.util.BaseTypeDefaultNull;
import top.zxybird.yun.common.util.Tuple;
import top.zxybird.yun.serialization.api.exception.DeserializationException;
import top.zxybird.yun.serialization.thrift.model.*;
import top.zxybird.yun.transport.api.TException;
import top.zxybird.yun.transport.api.TProtocolException;
import top.zxybird.yun.transport.api.TTransport;

public class ThriftBinaryDeserializer extends ThriftDeserializer {

  private static final long NO_LENGTH_LIMIT = -1;

  private TTransport trans_;

  public ThriftBinaryDeserializer(TTransport trans_) {
    this.trans_ = trans_;
  }

  /**
   * boolean类型需要4个字节
   * <p>
   * 1个字节的类型 2个字节的编号 1个字节的值
   */
  public static Tuple.Tuple2<Short, TBoolean> readTBoolean(TTransport trans_) throws TException {

    Short number = null;
    number = read(trans_, BaseTypeDefaultNull.SHORT);

    Boolean value = null;
    value = read(trans_, BaseTypeDefaultNull.BOOLEAN);

    TBoolean out = new TBoolean(value);

    return Tuple.tuple(number, out);

  }

  private static final Function<TTransport, TBoolean> deserializeTBoolean = (trans_) -> {

    Preconditions.checkNotNull(trans_);

    try {
      Short number = null;
      number = read(trans_, BaseTypeDefaultNull.SHORT);

      Boolean value = null;
      value = read(trans_, BaseTypeDefaultNull.BOOLEAN);

      TBoolean out = new TBoolean(value);
      out.setNumber(number);
      return out;
    } catch (TException e) {
      throw new DeserializationException();
    }

  };

  private TBoolean read(TBoolean type) throws TException {
    Short number = read(trans_, BaseTypeDefaultNull.SHORT);
    Boolean value = read(trans_, BaseTypeDefaultNull.BOOLEAN);
    TBoolean out = new TBoolean(value);
    out.setNumber(number);
    return out;
  }

  private Function readTBool = (type) -> {
    try {
      Short number = read(trans_, BaseTypeDefaultNull.SHORT);
      Boolean value = read(trans_, BaseTypeDefaultNull.BOOLEAN);
      TBoolean out = new TBoolean(value);
      out.setNumber(number);
      return out;
    } catch (Exception e) {
      return null;
    }
  };

  /**
   * byte类型需要4个字节
   * <p>
   * 1个字节的类型 2个字节的编号 1个字节的值
   */
  public static Tuple.Tuple2<Short, TByte> readTByte(TTransport trans_) throws TException {

    Short number = null;
    number = read(trans_, BaseTypeDefaultNull.SHORT);

    Byte value = null;
    value = read(trans_, value);

    TByte out = new TByte(value);

    return Tuple.tuple(number, out);
  }

  private TByte read(TByte type) throws TException {
    Short number = read(trans_, BaseTypeDefaultNull.SHORT);

    Byte value = read(trans_, BaseTypeDefaultNull.BYTE);

    TByte out = new TByte(value);
    out.setNumber(number);

    return out;
  }

  private static final Function<TTransport, TByte> deserializeTByte = (trans_) -> {

    Preconditions.checkNotNull(trans_);

    try {
      Short number = read(trans_, BaseTypeDefaultNull.SHORT);

      Byte value = read(trans_, BaseTypeDefaultNull.BYTE);

      TByte out = new TByte(value);
      out.setNumber(number);

      return out;
    } catch (TException e) {
      throw new DeserializationException();
    }

  };

  /**
   * double类型需要11个字节
   * <p>
   * 1个字节的类型 2个字节的编号 8个字节的值
   */
  public static Tuple.Tuple2<Short, TDouble> readTDouble(TTransport trans_) throws TException {
    Short number = null;
    number = read(trans_, BaseTypeDefaultNull.SHORT);

    Double value = null;
    value = read(trans_, BaseTypeDefaultNull.DOUBLE);

    TDouble out = new TDouble(value);

    return Tuple.tuple(number, out);
  }

  private TDouble read(TDouble type) throws TException {
    Short number = read(trans_, BaseTypeDefaultNull.SHORT);

    Double value = read(trans_, BaseTypeDefaultNull.DOUBLE);

    TDouble out = new TDouble(value);
    out.setNumber(number);

    return out;
  }

  /**
   * short类型需要5个字节
   * <p>
   * 1个字节的类型 2个字节的编号 2个字节的值
   */
  private static Tuple.Tuple2<Short, TShort> readTShort(TTransport trans_) throws TException {
    Short number = null;
    number = read(trans_, BaseTypeDefaultNull.SHORT);

    Short value = null;
    value = read(trans_, BaseTypeDefaultNull.SHORT);

    TShort out = new TShort(value);

    return Tuple.tuple(number, out);
  }

  private TShort read(TShort type) throws TException {
    Short number = read(trans_, BaseTypeDefaultNull.SHORT);

    Short value = read(trans_, BaseTypeDefaultNull.SHORT);

    TShort out = new TShort(value);
    out.setNumber(number);

    return out;
  }

  /**
   * int类型需要7个字节
   * <p>
   * 1个字节的类型 2个字节的编号 4个字节的值
   */
  private static Tuple.Tuple2<Short, TInt> readTInt(TTransport trans_) throws TException {
    Short number = null;
    number = read(trans_, BaseTypeDefaultNull.SHORT);

    Integer value = read(trans_, BaseTypeDefaultNull.INTEGER);

    TInt out = new TInt(value);

    return Tuple.tuple(number, out);
  }

  private TInt read(TInt type) throws TException {
    Short number = read(trans_, BaseTypeDefaultNull.SHORT);

    Integer value = read(trans_, BaseTypeDefaultNull.INTEGER);

    TInt out = new TInt(value);
    out.setNumber(number);

    return out;
  }

  /**
   * int类型需要11个字节
   * <p>
   * 1个字节的类型 2个字节的编号 8个字节的值
   */
  private static Tuple.Tuple2<Short, TLong> readTLong(TTransport trans_) throws TException {
    Short number = null;
    number = read(trans_, BaseTypeDefaultNull.SHORT);

    Long value = read(trans_, BaseTypeDefaultNull.LONG);

    TLong out = new TLong(value);

    return Tuple.tuple(number, out);
  }

  private TLong read(TLong type) throws TException {
    Short number = read(trans_, BaseTypeDefaultNull.SHORT);

    Long value = read(trans_, BaseTypeDefaultNull.LONG);

    TLong out = new TLong(value);
    out.setNumber(number);

    return out;
  }

  public static Tuple.Tuple2<Short, TString> readTString(TTransport trans_) throws TException {
    Short number = null;
    number = read(trans_, BaseTypeDefaultNull.SHORT);

    String value = read(trans_, BaseTypeDefaultNull.STRING);

    TString out = new TString(value);

    return Tuple.tuple(number, out);
  }

  private TString read(TString type) throws TException {
    Short number = read(trans_, BaseTypeDefaultNull.SHORT);

    String value = read(trans_, BaseTypeDefaultNull.STRING);

    TString out = new TString(value);
    out.setNumber(number);

    return out;
  }

  public static Tuple.Tuple2<Short, TStruct> readTStruct(TTransport trans_) throws TException {
    Short number = read(trans_, BaseTypeDefaultNull.SHORT);
    String value = read(trans_, BaseTypeDefaultNull.STRING);
    TString out = new TString(value);
    return null;
  }

  private TStruct read(TStruct type) throws TException {

    Short number = read(trans_, BaseTypeDefaultNull.SHORT);

    TStruct struct = new TStruct();
    while (true) {
      Byte typeId = read(trans_, BaseTypeDefaultNull.BYTE);

      if (typeId == TTypeConstant.STOP) {
        break;
      }

      Class clazz = TypeIdTTypeBiMap.getByTypeId(typeId);
      TType field = typeMap.get(clazz).function.apply(trans_);

      struct.addField(field);
    }

    struct.setNumber(number);

    return struct;
  }

  private static final Function<TTransport, TStruct> deserializeTStruct = (trans_) -> {

    Preconditions.checkNotNull(trans_);

    try {
      Short number = read(trans_, BaseTypeDefaultNull.SHORT);

      TStruct struct = new TStruct();
      while (true) {
        Byte typeId = read(trans_, BaseTypeDefaultNull.BYTE);

        if (typeId == TTypeConstant.STOP) {
          break;
        }

        Class clazz = TypeIdTTypeBiMap.getByTypeId(typeId);

        TType field = findByTypeWithDefault(clazz).function.apply(trans_);

        // typeMap.get(clazz).function.apply(trans_);

        // TType field = typeMap.get(clazz).apply(trans_);

        struct.addField(field);
      }

      struct.setNumber(number);

      return struct;
    } catch (TException e) {
      throw new DeserializationException();
    }

  };

  private enum Type {
    /**
     * Byte
     */
    _TByte(TByte.class, deserializeTBoolean),
    /**
     * Object
     */
    _TStruct(TStruct.class, deserializeTStruct);

    private Class clazz;
    private Function<TTransport, ? extends TType> function;

    Type(Class clazz, Function function) {
      this.clazz = clazz;
      this.function = function;
    }
  }

  private static Map<Class, Type> typeMap = Arrays.stream(Type.values())
      .collect(Collectors.toMap(x -> x.clazz, x -> x, (o1, o2) -> o1));

  private static Type findByTypeWithDefault(Class<?> clazz) {

    if (null == clazz) {
      return null;
    }

    Type type = typeMap.get(clazz);

    if (null != type) {
      return type;
    } else {
      return Type._TStruct;
    }

  }

  // private static Map<Class, Function<TTransport, ? extends TType>> typeMap = Maps.newHashMap();
  //
  // static {
  // typeMap.put(TBoolean.class, deserializeTBoolean);
  // }

  public static TMessage readTMessage(TTransport trans_) throws TException {

    TMessage message = new TMessage();

    Integer _i32 = read(trans_, BaseTypeDefaultNull.INTEGER);

    message.setVersion(_i32 & ThriftConstant.VERSION_MASK);

    message.setType((byte) (_i32 & ThriftConstant.TYPE_MASK));

    String name = read(trans_, BaseTypeDefaultNull.STRING);
    message.setName(name);

    message.setSeqid(read(trans_, BaseTypeDefaultNull.INTEGER));

    Byte typeId = null;

    Map<Short, TType> fieldMap = Maps.newHashMap();

    while (true) {
      typeId = read(trans_, typeId);

      if (typeId == TTypeConstant.STOP) {
        break;
      } else if (typeId == TTypeConstant.STRING) {
        Tuple.Tuple2<Short, TString> tuple = readTString(trans_);
        fieldMap.put(tuple._1(), tuple._2());
      }

    }

    List<Map.Entry<Short, TType>> fieldList = Lists.newArrayList(fieldMap.entrySet());

    fieldList.sort((o1, o2) -> {
      if (null == o1) {
        return -1;
      }
      if (null == o2) {
        return 1;
      }
      if (null == o1.getKey()) {
        return -1;
      }
      if (null == o2.getKey()) {
        return 1;
      }

      return o1.getKey() - o2.getKey();
    });

    message.setArgs(fieldList.stream().map(Map.Entry::getValue).collect(Collectors.toList()));

    return message;

  }

  // ======================================== 基本类型 ========================================

  public static Byte read(TTransport trans_, Byte out) throws TException {
    if (trans_.getBytesRemainingInBuffer() >= 1) {
      byte b = trans_.getBuffer()[trans_.getBufferPosition()];
      trans_.consumeBuffer(1);
      return b;
    }

    byte[] bin = new byte[1];
    trans_.readAll(bin, 0, 1);
    return bin[0];

  }

  public static Boolean read(TTransport _trans, Boolean out) throws TException {
    Byte _byte = null;
    _byte = read(_trans, _byte);
    return (_byte == 1);
  }

  public static Short read(TTransport trans_, Short out) throws TException {
    byte[] i16rd = new byte[2];
    byte[] buf = i16rd;
    int off = 0;

    if (trans_.getBytesRemainingInBuffer() >= 2) {
      buf = trans_.getBuffer();
      off = trans_.getBufferPosition();
      trans_.consumeBuffer(2);
    } else {
      trans_.readAll(i16rd, 0, 2);
    }

    return (short) (((buf[off] & 0xff) << 8) | ((buf[off + 1] & 0xff)));
  }

  public static Integer read(TTransport trans_, Integer out) throws TException {
    byte[] i32rd = new byte[4];
    byte[] buf = i32rd;
    int off = 0;

    if (trans_.getBytesRemainingInBuffer() >= 4) {
      buf = trans_.getBuffer();
      off = trans_.getBufferPosition();
      trans_.consumeBuffer(4);
    } else {
      trans_.readAll(i32rd, 0, 4);
    }
    return ((buf[off] & 0xff) << 24) | ((buf[off + 1] & 0xff) << 16) | ((buf[off + 2] & 0xff) << 8)
        | ((buf[off + 3] & 0xff));
  }

  public static Long read(TTransport trans_, Long out) throws TException {
    byte[] i64rd = new byte[8];
    byte[] buf = i64rd;
    int off = 0;

    if (trans_.getBytesRemainingInBuffer() >= 8) {
      buf = trans_.getBuffer();
      off = trans_.getBufferPosition();
      trans_.consumeBuffer(8);
    } else {
      trans_.readAll(i64rd, 0, 8);
    }

    return ((long) (buf[off] & 0xff) << 56) | ((long) (buf[off + 1] & 0xff) << 48)
        | ((long) (buf[off + 2] & 0xff) << 40) | ((long) (buf[off + 3] & 0xff) << 32)
        | ((long) (buf[off + 4] & 0xff) << 24) | ((long) (buf[off + 5] & 0xff) << 16)
        | ((long) (buf[off + 6] & 0xff) << 8) | ((long) (buf[off + 7] & 0xff));
  }

  public static Double read(TTransport trans_, Double out) throws TException {
    return Double.longBitsToDouble(read(trans_, BaseTypeDefaultNull.LONG));
  }

  public static String read(TTransport trans_, String out) throws TException {
    Integer _int = null;

    int size = read(trans_, _int);

    if (size < 0) {
      throw new TProtocolException(TProtocolException.NEGATIVE_SIZE, "Negative length: " + size);
    }

    if (NO_LENGTH_LIMIT > 0 && size > NO_LENGTH_LIMIT) {
      throw new TProtocolException(TProtocolException.SIZE_LIMIT, "String field exceeded string size limit");
    }

    if (trans_.getBytesRemainingInBuffer() >= size) {
      try {
        String s = new String(trans_.getBuffer(), trans_.getBufferPosition(), size, "UTF-8");
        trans_.consumeBuffer(size);
        return s;
      } catch (UnsupportedEncodingException e) {
        throw new TException("JVM DOES NOT SUPPORT UTF-8");
      }
    }

    try {
      byte[] buf = new byte[size];
      trans_.readAll(buf, 0, size);
      return new String(buf, "UTF-8");
    } catch (UnsupportedEncodingException uex) {
      throw new TException("JVM DOES NOT SUPPORT UTF-8");
    }

  }

}
